package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DwdTrafficSplitApp {


    //1   从kafka中读取数据 ，转为stream
    //2   把jsonstring 转为jsonObject
    //3   利用状态校正标识
    //4   拆分jsonobject 五种流数据 ，打outputtag做侧输出
    //5   分别写入不同的kafka主题
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(3000);
        //1   从kafka中读取数据 ，转为stream
        String topic="topic_log";
        String groupId="dwd_traffic_split_app";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStream = env.addSource(kafkaConsumer);

        //2   把jsonstring 转为jsonObject
         SingleOutputStreamOperator<JSONObject> jsonObjStream = kafkaStream.map(jsonString -> JSON.parseObject(jsonString));

//
//        //状态 ： 算子状态  、键控状态
//
//        //3   利用状态校正标识 is_new  业务上的状态数据  而且 数据量有可能较大
//        //  使用的是算子状态还是键控状态
//        // 就看流是否被keyby
//        //  1 定义 以key为的单位 描述符 2 初始化 open方法 3 提取 getRuntimeContext().getState(描述符) 4 使用 state.value 查    state.update 写
//        // is_new =1 前端清理数据 会造成标识丢失 错判为新访客
//        // 新访客的认定范围 ： 当日是用户的首次访问， 该日此访客都被认定为新访客
//        //  1  状态不存在 ， 认定为新访客， 当日日期作为该访客的首次访问时间 保存在状态中
//        //  2  状态在 【首次访问时间】 和【当前访问时间】 比较 如果二者相同 ，依旧认定为新访客
//        //  3  状态在 【首次访问时间】 和【当前访问时间】  二者不同，认定为旧访客
//
//        // is_new =0 前端认定为老访客 基本正确不用校正
//        //  1 状态在   不管
//        //  2 状态不在  以前端标识为准   补充状态 【首次访问时间】 置为【当前访问时间】前一日
//
//
        KeyedStream<JSONObject, String> midKeybyStream = jsonObjStream.keyBy(jsonobj -> jsonobj.getJSONObject("common").getString("mid"));
//
//
        SingleOutputStreamOperator<JSONObject> jsonObjAfterCheckStream = midKeybyStream.map(new RichMapFunction<JSONObject, JSONObject>() {
            //键控状态 把每个状态以key为单位存储，所以每个key对应一个单值即可。
            ValueStateDescriptor<String> firstVisitDateStateDesc = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                firstVisitDateStateDesc = new ValueStateDescriptor("first_visit_date_state", String.class);
                firstVisitDateStateDesc.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1))
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    .build());
            }

            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {
                ValueState<String> firstVisitDtState = getRuntimeContext().getState(firstVisitDateStateDesc);
                String firstVisitDt = firstVisitDtState.value();
                //     firstVisitDtState.update(firstVisitDt);

                String isNewByApp = jsonObject.getJSONObject("common").getString("is_new");
                Long ts = jsonObject.getLong("ts");

                String curVisitDt = DateFormatUtil.toDate(ts);
                // is_new =1 前端清理数据 会造成标识丢失 错判为新访客
                if (isNewByApp.equals("1")) {
                    //  1  状态不存在 ， 认定为新访客， 当日日期作为该访客的首次访问时间 保存在状态中
                    if (firstVisitDt == null || firstVisitDt.length() == 0) {
                        firstVisitDtState.update(curVisitDt);
                    } else {
                        //  2  状态在 【首次访问时间】 和【当前访问时间】 比较 如果二者相同 ，依旧认定为新访客
                        //  3  状态在 【首次访问时间】 和【当前访问时间】  二者不同，认定为旧访客
                        if (!firstVisitDt.equals(curVisitDt)) {
                            jsonObject.getJSONObject("common").put("is_new", "0");
                        }
                    }

                } else {
                    // is_new =0 前端认定为老访客 基本正确不用校正
                    //  1 状态在   不管
                    //  2 状态不在  以前端标识为准   补充状态 【首次访问时间】 置为【当前访问时间】前一日
                    if (firstVisitDt == null || firstVisitDt.length() == 0) {
                        long yesterdayTs = ts - (1000 * 3600 * 24);
                        String yesterdayDt = DateFormatUtil.toDate(yesterdayTs);
                        firstVisitDtState.update(yesterdayDt);
                    }
                }


                return jsonObject;
            }
        });
 //       jsonObjAfterCheckStream.print();
//
//        //4   拆分jsonobject 五种流数据 ，打outputtag做侧输出    页面  启动  动作 曝光 报错
        //侧输出的步骤
        // 4.1 定义侧输出的标签outputtag   4个侧输出tag
        //4.2 通过代码根据流中元素进程判断 组织侧输出的结构 打tag    processFunction
        //4.3 在通过标签把侧输出的流拆分出来  Sideoutput  多个stream
        OutputTag<JSONObject> startupTag= new OutputTag<JSONObject>("startup"){};
        OutputTag<JSONObject> actionTag= new OutputTag<JSONObject>("action"){};
        OutputTag<JSONObject> displayTag= new OutputTag<JSONObject>("display"){};
        OutputTag<JSONObject> errorTag= new OutputTag<JSONObject>("error"){};
        SingleOutputStreamOperator<JSONObject> pageStream = jsonObjAfterCheckStream.process(new ProcessFunction<JSONObject, JSONObject>() {
            @Override
            public void processElement(JSONObject jsonObject, Context ctx, Collector<JSONObject> out) throws Exception {
                // ctx.output(outputTag,"xxxx"); 标记为某个侧输出    启动  动作 曝光 报错
                //  out.collect(xx);  保留在主流中     页面

                // 1  判断是否有报错  如果有直接标识为报错数据  报错数据保持原值  不做处理
                // 2  判断是否为启动日志 依据只要有start 标识为启动 , 不做特别处理
                // 3  page操作中会有 多个displays  要拆分成多条  每条包含 page common display ts 标识为曝光
                // 4  page操作中会有 多个action  要拆分成多条  每条包含 page common action ts   标识为动作
                // 5  page 去除掉 displays actions   其他保留   保留在主流中
                if (jsonObject.getJSONObject("err") != null && jsonObject.getJSONObject("err").size() > 0) {
                    ctx.output(errorTag, jsonObject);
                } else if (jsonObject.getJSONObject("start") != null && jsonObject.getJSONObject("start").size() > 0) {
                    ctx.output(startupTag, jsonObject);
                } else { //页面
                    // 3  page操作中会有 多个displays  要拆分成多条  每条包含 page common display ts 标识为曝光
                    JSONArray displayJsonArr = jsonObject.getJSONArray("displays");
                    if (displayJsonArr != null && displayJsonArr.size() > 0) {
                        for (int i = 0; i < displayJsonArr.size(); i++) {
                            JSONObject displayAllJsonObj = new JSONObject();
                            JSONObject displayJsonObj = displayJsonArr.getJSONObject(i);
                            displayAllJsonObj.put("common", jsonObject.getJSONObject("common"));
                            displayAllJsonObj.put("page", jsonObject.getJSONObject("page"));
                            displayAllJsonObj.put("display", displayJsonObj);
                            displayAllJsonObj.put("ts", jsonObject.getLong("ts"));
                            ctx.output(displayTag, displayAllJsonObj);
                        }
                    }
                    // 4  page操作中会有 多个action  要拆分成多条  每条包含 page common action ts   标识为动作
                    JSONArray actionJsonArr = jsonObject.getJSONArray("actions");
                    if (actionJsonArr != null && actionJsonArr.size() > 0) {
                        for (int i = 0; i < actionJsonArr.size(); i++) {
                            JSONObject actionAllJsonObj = new JSONObject();
                            JSONObject actionJsonObj = actionJsonArr.getJSONObject(i);
                            actionAllJsonObj.put("common", jsonObject.getJSONObject("common"));
                            actionAllJsonObj.put("page", jsonObject.getJSONObject("page"));
                            actionAllJsonObj.put("action", actionJsonObj);
                            actionAllJsonObj.put("ts", jsonObject.getLong("ts"));
                            ctx.output(actionTag, actionAllJsonObj);
                        }
                    }
                    // 5  page 去除掉 displays actions   其他保留   保留在主流中
                    jsonObject.remove("displays");
                    jsonObject.remove("actions");
                    out.collect(jsonObject);

                }

            }
        });
        //4.3 在通过标签把侧输出的流拆分出来  Sideoutput  多个stream
        DataStream<JSONObject> startupStream = pageStream.getSideOutput(startupTag);
        DataStream<JSONObject> errorStream = pageStream.getSideOutput(errorTag);
        DataStream<JSONObject> displayStream = pageStream.getSideOutput(displayTag);
        DataStream<JSONObject> actionStream = pageStream.getSideOutput(actionTag);


       pageStream.print("页面流：：：：");
       startupStream.print("启动流：：：：");
       errorStream.print("报错流：：：：");
       displayStream.print("曝光流：：：：");
       actionStream.print("动作流：：：：");

//        //5   分别写入不同的kafka主题

        actionStream.map(jsonObj->jsonObj.toJSONString()).addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_action_log"));
        displayStream.map(jsonObj->jsonObj.toJSONString()).addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_display_log"));
        startupStream.map(jsonObj->jsonObj.toJSONString()).addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_start_log"));
        pageStream.map(jsonObj->jsonObj.toJSONString()).addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_page_log"));
        errorStream.map(jsonObj->jsonObj.toJSONString()).addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_error_log"));

        env.execute();
    }
}
